package com.dys.flink.wc1.netty.handler;

import com.dys.flink.wc1.model.Product;
import com.dys.flink.wc1.util.JsonUtils;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;

/**
 * 帧格式： 4字节长度（大端） + JSON
 */
public class ProductDecoder extends LengthFieldBasedFrameDecoder {

    private static final Logger logger = LoggerFactory.getLogger(ProductDecoder.class);

    public ProductDecoder() {
        super(16 * 1024 * 1024,   // maxFrameLength
              0,                   // lengthFieldOffset
              4,                   // lengthFieldLength
              0,                   // lengthAdjustment
              4);                  // initialBytesToStrip（去掉长度字段本身）
    }

    @Override
    protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        ByteBuf frame = (ByteBuf) super.decode(ctx, in);
        if (frame == null) {          // 数据不够
            return null;
        }
        try {
            String json = frame.toString(StandardCharsets.UTF_8);
            Product product = JsonUtils.fromJson(json, Product.class);
            logger.debug("Successfully decoded product: {}", product);
            return product;
        } finally {
            frame.release();          // 释放临时帧
        }
    }
}